package com.guandata.cdc.source.impl;

import com.guandata.cdc.helper.ParamsHelper;
import com.guandata.cdc.model.source.ConnectOptionsLoader;
import com.guandata.cdc.model.source.SourceConnectParam;
import com.guandata.cdc.model.DataTableType;
import com.guandata.cdc.source.GuanDataSourceFactory;
import com.guandata.cdc.model.db_type.mysql.MysqlSourceConfig;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.connector.jdbc.mysql.database.catalog.MySqlCatalog;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.ObjectPath;

import java.lang.reflect.InvocationTargetException;

/**
 * <pre>
 * Setup:
 *     CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';
 *     GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';
 * </pre>
 * @author daishuyuan
 */
@Slf4j
public class MysqlSQLSourceFactory extends GuanDataSourceFactory<MysqlSourceConfig> {
     static {
         try {
             // https://stackoverflow.com/questions/63119874/using-mysql-connector-java-in-flink-no-suitable-driver-found-when-connecting-to
             Class.forName("com.mysql.jdbc.Driver").getDeclaredConstructor().newInstance();
             Class.forName("com.mysql.cj.jdbc.Driver").getDeclaredConstructor().newInstance();
         } catch (InstantiationException | IllegalAccessException | ClassNotFoundException | InvocationTargetException |
                  NoSuchMethodException e) {
             log.error("MysqlSQLSourceFactory init error", e);
         }
     }

    public MysqlSQLSourceFactory(MysqlSourceConfig config) {
       super(config);
       super.schema = initializeSchema();
    }

    @SneakyThrows
    private Schema initializeSchema() {
        var jdbcCatalog = new MySqlCatalog(
            Thread.currentThread().getContextClassLoader(),
            config.getCatalogName("mysql_source"),
            config.sourceDatabase,
            config.sourceUsername,
            config.sourcePassword,
            config.sourceJdbcUrl
        );
       return jdbcCatalog
           .getTable(ObjectPath.fromString(config.sourceTable))
           .getUnresolvedSchema();
    }

    @SneakyThrows
    @Override
    public Table registerSourceTable(StreamTableEnvironment tableEnv) {
        var sourceFlinkTableName = ParamsHelper.getInnerTableName(config.rawSourceTable);
        var descriptorBuilder = TableDescriptor.forConnector(config.connector).schema(this.schema);

        log.info("{} table's schema: {}", sourceFlinkTableName, this.schema);

        ConnectOptionsLoader
            .<SourceConnectParam>builder(DataTableType.MYSQL, descriptorBuilder)
            .add(SourceConnectParam.HOST_NAME, config.sourceHostname)
            .add(SourceConnectParam.PORT, String.valueOf(config.sourcePort))
            .add(SourceConnectParam.DATABASE_NAME, config.sourceDatabase)
            .add(SourceConnectParam.TABLE_NAME, config.rawSourceTable)
            .add(SourceConnectParam.USERNAME, config.sourceUsername)
            .add(SourceConnectParam.PASSWORD, config.sourcePassword)
            .add(SourceConnectParam.SCAN_STARTUP_MODE, config.startUpMode.getCode())
            .add(SourceConnectParam.SERVER_ID, config.serverId);
        tableEnv.createTable(sourceFlinkTableName, descriptorBuilder.build());
        return dropColumnsFrom(tableEnv.from(sourceFlinkTableName));
    }

}
